package com.mzj.eagle.pool.impl;

import static redis.clients.jedis.Protocol.toByteArray;
import static redis.clients.jedis.Protocol.Command.AUTH;
import static redis.clients.jedis.Protocol.Command.SELECT;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.moilioncircle.redis.replicator.Configuration;
import com.moilioncircle.redis.replicator.RedisURI;
import com.moilioncircle.redis.replicator.rdb.datatype.DB;
import com.moilioncircle.redis.replicator.rdb.datatype.KeyValuePair;
import com.moilioncircle.redis.replicator.rdb.dump.datatype.DumpKeyValuePair;
import com.mzj.eagle.pool.IRedisClientPool;
import com.mzj.eagle.redis.RedisClient;

import redis.clients.jedis.Protocol.Command;

public class RedisClientPoolImpl implements IRedisClientPool{
	
	private static final Logger LOG = Logger.getLogger(RedisClientPoolImpl.class);  

	private int times = 5;
	
	private String host;
	
	private Integer port;
	
	private RedisURI uri;
	
	private byte[][] dbArgs;
	
	private String threadName = Thread.currentThread().getName();
	
	// 空闲连接  
    private RedisClient client = null;  
	
	public RedisClientPoolImpl(RedisURI uri) {  
        super();  
        this.host = uri.getHost();
        this.port = uri.getPort();
        this.uri = uri;
        init();
    } 
	
	 // 初始化  
    public void init() {  
        this.client = newConnection();
    }  
    
    private void tempSendKv(RedisClient redisClient, AtomicInteger dbnum, KeyValuePair<?> kv) throws Exception{
    	DB db = kv.getDb();
        int index = -1;
        if (db != null && (index = (int) db.getDbNumber()) != dbnum.get()) {
        	redisClient.send(SELECT, toByteArray(index));
            dbnum.set(index);
            LOG.info(Thread.currentThread().getName() + " SELECT:" + index);
        }

        DumpKeyValuePair mkv = (DumpKeyValuePair) kv;
        // Step2: restore dump data
        if (mkv.getExpiredMs() == null) {
            Object r = redisClient.restore(mkv.getRawKey(), 0L, mkv.getValue(), false);
            //	LOG.info(Thread.currentThread().getName() +new String(mkv.getRawKey(), UTF_8)+ new String(mkv.getValue(), UTF_8) +" " +r);
            LOG.info(Thread.currentThread().getName() +" result:" +r);
        } else {
            long ms = mkv.getExpiredMs() - System.currentTimeMillis();
            if (ms <= 0) return;
            Object r = redisClient.restore(mkv.getRawKey(), ms, mkv.getValue(), false);
            LOG.info(Thread.currentThread().getName() +" result:" +r);
        }
        
    }
	
	public void sendKv(AtomicInteger dbnum, KeyValuePair<?> kv){
		for(int i=0; i<times; i++ ) {
			boolean successFlag = false;
			RedisClient redisClient = this.getConnection();
			try {
				this.tempSendKv(redisClient,dbnum, kv);
				successFlag = true;
			} catch (Exception e) {
				 LOG.error(Thread.currentThread().getName() +" " +e.getMessage(),e);
			}
			if(successFlag) {
				return;
			}
		}
	}
	
	public void sendCommand(byte[] command, byte[][] args) {
		for(int i=0; i<times; i++ ) {
			boolean successFlag = false;
			RedisClient redisClient = this.getConnection();
			try {
				if(this.dbArgs != null) {
					redisClient.send(Command.SELECT.raw, dbArgs);
				}
				Object r = redisClient.send(command, args);
				if(Arrays.equals(command, Command.SELECT.raw)) { //命令执行成功 保存selectDB
					this.dbArgs = args;
				}
				LOG.info(Thread.currentThread().getName() +" result:" +r);
				successFlag = true;
			} catch (Exception e) {
				 LOG.error(Thread.currentThread().getName() +" " +e.getMessage(),e);
			}
			if(successFlag) {
				return;
			}
		}
	}

	public RedisClient getConnection() {
		if(!this.isActive(client)) {
			client = newConnection();
		}
		return client;
	}

	public void destroy() {
		if(this.isActive(client)) {
			client.close();
			client = null;
		}
	}

	
	// 获得新连接  
    private synchronized RedisClient newConnection() {  
    	RedisClient conn = new RedisClient(host, port);
    	conn.setSoTimeout(30000);
    	conn.setConnectionTimeout(30000);
    	Configuration tconfig = Configuration.valueOf(uri);
        if (tconfig.getAuthPassword() != null) {
            Object auth = conn.send(AUTH, tconfig.getAuthPassword().getBytes());
            LOG.info(Thread.currentThread().getName() + " AUTH:" + auth);
        }
        return conn;  
    }  
    
	public boolean isActive(RedisClient client) {
        if (client == null) {  
            return false;  
        }else {
        	return client.isConnected();
        }
	}

}
